package org.onion.pattern.event.consumer.log;


import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public abstract class AsyncEventConsumeLogStorage implements EventConsumeLogStorage {

    private final LinkedBlockingQueue<EventConsumeLog> queue;
    private final ExecutorService                      executorService;
    private final AtomicBoolean                        isClosed;
    private final AtomicBoolean                        started;
    private final int                                  queueSize;
    private final int                                  batchStoreSize;
    private final int                                  maxDelay;

    public AsyncEventConsumeLogStorage(int queueSize, int batchStoreSize, int maxDelay) {
        //maxDelay 0.5秒
        //batchStoreSize 一次保存 50 条
        //queueSize 队列大小 5000 条
        this.batchStoreSize = batchStoreSize;
        this.queueSize = queueSize;
        this.maxDelay = maxDelay;
//        this.threadSize = threadSize;
        this.queue = new LinkedBlockingQueue<>(queueSize);
        this.executorService = Executors.newSingleThreadExecutor();
        this.isClosed = new AtomicBoolean(false);
        this.started = new AtomicBoolean(false);
        //        this.dingTalkClientCache = CacheBuilder.newBuilder()
//                //配置最大容量为100，基于容量进行回收
//                .maximumSize(200)
//                //配置写入后多久使缓存过期
//                .expireAfterWrite(3, TimeUnit.HOURS)
//                //多久没有访问之后过期
////                .expireAfterAccess(10, TimeUnit.MINUTES)
//                //配置写入后多久刷新缓存
////                .refreshAfterWrite(3, TimeUnit.SECONDS)
//                //key使用弱引用-WeakReference
//                .weakKeys()
//                //当Entry被移除时的监听器-下文会讲述
////                .removalListener(notification -> System.out.println("notification=" + notification))
//                //创建一个CacheLoader，重写load方法，以实现"当get时缓存不存在，则load，放到缓存并返回的效果
//                .build();
    }


    protected abstract void batchSave(List<EventConsumeLog> consumeLogs);

    @Override
    public final void save(EventConsumeLog consumeLog) {

        if (!this.started.get()) {
            startExecutor();
        }

        try {
            if (queue.size() >= this.queueSize) {
                log.warn("【消费日志】 消息溢出已丢弃,队列大小 {}", this.queueSize);
                this.batchSave(Collections.singletonList(consumeLog));     //直接保存
                return;
            }

            if (this.isClosed.get()) { //已经关闭了
                this.batchSave(Collections.singletonList(consumeLog));     //直接保存
            }

            queue.put(consumeLog);
        } catch (Exception e) {
            log.warn("【消费日志】", e);
        }
    }

    public void close() {
        this.isClosed.set(true);

        try {
            this.executorService.shutdownNow();
        } catch (Exception e) {
        }

        try {
            int size = this.queue.size();
            if (size > 0) {
                log.error("【消费日志】 容器将要关闭,还有 {} 条事件未处理", size);
            }
            this.queue.clear();
        } catch (Exception e) {
        }
    }


    private synchronized void startExecutor() {
        if (this.started.get()) {
            return;
        }

        //启动线程池
        this.executorService.submit(this::exec);

        this.started.set(true);
    }

    private void exec() {

        final List<EventConsumeLog> entities = new ArrayList<>(this.batchStoreSize);

        while (!isClosed.get() && !Thread.interrupted()) {
            try {
                long currentTs = System.currentTimeMillis();
                EventConsumeLog consumeLog = this.queue.poll(this.maxDelay, TimeUnit.MILLISECONDS);
                if (consumeLog == null) {
                    continue;
                }

                entities.add(consumeLog);
                queue.drainTo(entities, this.batchStoreSize - 1); //批量转移, 将队列中最多this.batchStoreSize - 1个元素转移到entities中
                boolean fullPack = entities.size() == this.batchStoreSize;
                if (log.isDebugEnabled()) {
                    log.debug("EventConsumeLog save {} entities", entities.size());
                    log.trace("EventConsumeLog save entities: {}", entities);
                }

                // 消息过期处理 ， 消息过滤 ， 排序？？
                this.batchSave(entities);
//
                if (!fullPack) {
                    long remainingDelay = this.maxDelay - (System.currentTimeMillis() - currentTs);
                    if (remainingDelay > 0) {
                        Thread.sleep(remainingDelay);
                    }
                }

            } catch (Throwable e) {
                if (e instanceof InterruptedException) {
                    log.warn("【消费日志】 Queue polling was interrupted");
                    Thread.currentThread().interrupt();
                    break;
                } else {
                    log.error("【消费日志】 消息保存失败 {} entities", entities.size(), e);
                }
            } finally {
                entities.clear();
            }
        }

    }
}
